package com.tiger.cloud.common.redis.queue;

import java.io.IOException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

import org.apache.log4j.Logger;

import com.tiger.cloud.util.ThreadUtils;

public class RedisMessageConsumer {

	private static Logger log = Logger.getLogger(RedisMessageConsumer.class);

	private RedisQueue redisQueue;
	private MessageListener listener;
	private ExecutorService exec = null;
	private AtomicBoolean started = new AtomicBoolean(false);
	private AtomicBoolean closed = new AtomicBoolean(false);

	public RedisMessageConsumer(RedisQueue redisQueue, MessageListener listener) {
		this.redisQueue = redisQueue;
		this.listener = listener;
	}

	private String getMessage(int timeOut) {
		return redisQueue.askForMessage(timeOut);
	}

	public void consume() {
		String message = getMessage(2); // brpoplpush 2s
		try {
			if (message != null && message.length() > 0) {
				listener.onMessage(message);
				redisQueue.onCompleted(message);
			} else {
				Thread.sleep(5000);
			}
		} catch (Exception e) {
			log.error(e.getMessage(), e);
		}
	}

	public void start() {
		if (!started.compareAndSet(false, true)) {
			throw new IllegalStateException("this consumer already start!");
		}
		if (exec == null) {
			exec = Executors.newSingleThreadExecutor();
		}
		exec.submit(new Runnable() {
			@Override
			public void run() {
				while (!closed.get()) {
					consume();
				}
			}
		});
	}

	public void close() throws IOException {
		if (!closed.compareAndSet(false, true)) {
			throw new IllegalStateException("already close!");
		}
		if (exec != null) {
			ThreadUtils.shutdownAndAwaitTermination(exec, 10, TimeUnit.SECONDS);
		}
	}
}
